package com.bjc.controller;

import com.bjc.mqtt.enums.QosEnum;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.boot.web.client.RestTemplateBuilder;
import org.springframework.http.HttpEntity;
import org.springframework.http.HttpHeaders;
import org.springframework.http.MediaType;
import org.springframework.http.ResponseEntity;
import org.springframework.web.bind.annotation.*;
import org.springframework.web.client.RestTemplate;

import java.util.HashMap;
import java.util.Map;

/**
 * @program: emq-demo
 * @description: 通过EMQX 的webhook将客户端的连接断开等事件通知到我们自建的服务上，通过事件类型获取客户端的连 接状态，然后将客户端的连接状态进行存储，并且提供HTTP API供后台系统查询所有客户端的状态
 * @author: bjc
 * @create: 2021-08-28 20:57
 */
@RequestMapping("/mqtt")
@RestController
public class WebHookController {

    private static final Logger LOG = LoggerFactory.getLogger(WebHookController.class);

    private Map<String, Boolean> clientStatusMap = new HashMap<>();

    @PostMapping("/webhook")
    public void hook(@RequestBody Map<String,Object> params){
        LOG.info("emqx 触发 webhook,请求体数据={}",params);
        String action = (String) params.get("action");
        String clientid = (String) params.get("clientid");
        if (action.equals("client_connected")){
            //客户端成功接入
            LOG.info("客户端{}接入本系统",clientid);
            clientStatusMap.put(clientid,true);
            //自动位客户端产生订阅
            autoSub(clientid,"autoSub/#",QosEnum.QOS2,true);
        }
        if (action.equals("client_disconnected")){
            //客户端断开连接
            LOG.info("客户端{}下线",clientid);
            clientStatusMap.put(clientid,false);
            //自动的为客户端取消订阅
            autoSub(clientid,"autoSub/#",QosEnum.QOS2,false);
        }
    }

    @GetMapping("/allStatus")
    public Map getAllStatus(){
        return clientStatusMap;
    }

    /**
     * 自动订阅或取消订阅
     * @param clientId
     * @param topicFilter
     * @param qos
     * @param sub
     */
    private void autoSub(String clientId,String topicFilter, QosEnum qos,boolean sub){
        RestTemplate restTemplate = new RestTemplateBuilder()
                .basicAuthentication("admin", "public")
                .defaultHeader(HttpHeaders.CONTENT_TYPE,
                        org.springframework.http.MediaType.APPLICATION_JSON_VALUE)
                .build();
//装配参数
        Map param = new HashMap();
        param.put("clientid",clientId);
        param.put("qos",qos.value());
        param.put("topic",topicFilter);
        LOG.info("请求emq的相关参数:{}",param);
        HttpHeaders headers = new HttpHeaders();
        headers.setContentType(org.springframework.http.MediaType.APPLICATION_JSON);
        HttpEntity<Object> entity = new HttpEntity<Object>(param,headers);
//自动订阅
        if(sub){
            new Thread(()->{
                ResponseEntity<String> responseEntity =
                        restTemplate.postForEntity("http://192.168.242.97:8081/api/v4/mqtt/subscribe", entity,
                                String.class);
                LOG.info("自动订阅的结果:{}",responseEntity.getBody());
            }).start();
            return;
        }
//自动取消订阅
        ResponseEntity<String> responseEntity =
                restTemplate.postForEntity("http://192.168.242.97:8081/api/v4/mqtt/unsubscribe", entity,
                        String.class);
        LOG.info("自动取消订阅的结果:{}",responseEntity.getBody());
    }
}
